#include "HBaseClient/HBaseClient.h"
#include "ElasticsearchClient/ElasticsearchClient.h"
#include "HdfsClient/HdfsClient.h"
#include "fstream"
#include "sstream"

using namespace std;
using namespace rapidjson;
using namespace httplib;


//#define hbase
//#define hdfs
//#define es
#define system
//#define reset

string GetStr(const string& tableName, const list<tuple<string, list<tuple<string, string, bool, bool>>>>& content)
{
    auto dom = Document();
    dom.SetObject();
    dom.AddMember("tableName", Document().SetString(tableName.c_str(), tableName.size(), dom.GetAllocator()), dom.GetAllocator());
    dom.AddMember("rows", Document().SetArray(), dom.GetAllocator());
    auto rowsDom = dom["rows"].GetArray();
    for (auto& i : content)
    {
        auto& rowKey = get<0>(i);
        auto& columns = get<1>(i);
        auto rowDom = Document();
        rowDom.SetObject();
        rowDom.AddMember("rowKey", Document().SetString(rowKey.c_str(), rowKey.size(), dom.GetAllocator()), dom.GetAllocator());
        rowDom.AddMember("columns", Document().SetArray(), dom.GetAllocator());
        auto columnsDom = rowDom["columns"].GetArray();
        for (auto& j : columns)
        {
            auto& column = get<0>(j);
            auto& value = get<1>(j);
            auto& canSearch = get<2>(j);
            auto& canCut = get<3>(j);
            auto columnDom = Document();
            columnDom.SetObject();
            columnDom.AddMember("column", Document().SetString(column.c_str(), column.size(), dom.GetAllocator()), dom.GetAllocator());
            columnDom.AddMember("value", Document().SetString(value.c_str(), value.size(), dom.GetAllocator()), dom.GetAllocator());
            columnDom.AddMember("canSearch", Document().SetBool(canSearch), dom.GetAllocator());
            columnDom.AddMember("canCut", Document().SetBool(canCut), dom.GetAllocator());
            columnsDom.PushBack(columnDom, dom.GetAllocator());
        }
        rowsDom.PushBack(rowDom, dom.GetAllocator());
    }

    return RemoveRedundancyChars(GetJsonString(dom));
}

int main()
{
#ifdef hbase
    auto client = HBaseClient("192.168.56.51", 8080);
    cout << client.CreateTable(TableSchema("test_table", { tuple("_doc", 1) })) << endl;
    cout << client.PutCells("new_table", RowsShell({Row("1", {Cell("_doc:Name", "Castle")})})) << endl;
    auto result = client.GetRow("new_table", "1");
    if (result.has_value())
    {
        cout << result.value().Key << ":" << endl;
        for (auto& i : result.value().Cells)
            cout << "-\t" << i.column << "\t" << i.value << endl;
    }
    cout << client.DeleteRow("new_table", "1") << endl;
    return 0;
#elif defined(hdfs)
    auto client = HdfsClient("192.168.56.51", 14000);
    cout << client.UploadFile("/MainWidget.ui", "/home/castle/MainWidget.ui") << endl;
    cout << client.DownloadFile("/MainWidget.ui", "/home/castle/tmp/MainWidget.ui") << endl;
    cout << client.DeleteFile("/MainWidget.ui") << endl;
#elif defined(es)
    auto client = ElasticsearchClient("192.168.56.51", 9200);
    cout << client.CreateIndex("test_table") << endl;
    cout << client.CreateDocument("test_table", "1", "Name", "Castle", false) << endl;
    usleep(400000);
    auto searchResult = client.SearchInIndex("test_table", "Castle");
    if (searchResult.has_value())
        for (auto& i : searchResult.value())
            cout << get<0>(i) << " " << get<1>(i) << " " << get<2>(i) << endl;
    else
        cout << "Failed!" << endl;
    cout << client.DeleteDocument("test_table", "1", "Name") << endl;
    cout << client.DeleteIndex("test_table");
    return 0;
#elif defined(system)
    string tableName = "person";
    ifstream ifs;
    ifs.open("../../DataGen/persons.txt", ios::in);
    if (!ifs.is_open())
        return 1;
    list<string> ls;
    while (true)
    {
        string s;
        if (ifs.peek() == '\n')
            break;
        ifs >> s;
        ls.emplace_back(s);
    }

    list<tuple<string, list<tuple<string, string, bool, bool>>>> dataList;
    while (true)
    {
        if (ifs.eof())
            break;
        string rowKey;
        list<tuple<string, string, bool, bool>> innerDataList({});
        for (auto& i : ls)
        {
            if (i == "Uid")
            {
                ifs >> rowKey;
            }
            else
            {
                string tmp;
                ifs >> tmp;
                bool canSearch = true;
                bool canCut = true;
                innerDataList.emplace_back(tie(i, tmp, canSearch, canCut));
            }
        }
        dataList.emplace_back(tie(rowKey, innerDataList));
    }

    ifs.close();

    auto body = GetStr(tableName, dataList);

    auto path = stringstream();
    path << "/manage/" << tableName;
    Headers headers;
    headers.insert(make_pair("Accept", "application/json"));
    Client client("192.168.56.51", 1234);
    client.set_connection_timeout(100);
    client.set_read_timeout(100);
    client.set_write_timeout(100);
    bool ok = false;
    Timer t;
    auto resp = client.Put(path.str().c_str(), headers, body, "application/json");
    auto time = t.GetMSec();
    do
    {
        if (!resp)
        {
            ok = false;
            break;
        }
        if (resp->status != 200)
        {
            ok = false;
            break;
        }
        ok = true;
    } while (false);
    cout << ok << " " << time << "ms" << endl;
    if (!ok && resp)
        cout << resp->body << endl;
    return 0;

#elif defined(reset)
    auto hdfsClient = HdfsClient("192.168.56.51", 14000);
    auto elasticSearchClient = ElasticsearchClient("192.168.56.51", 9200);
    auto hbaseClient = HBaseClient("192.168.56.51", 8080);

    auto tables = hbaseClient.GetTables();
    if (tables.has_value())
        for (auto& i : tables.value())
            hbaseClient.DeleteTable(i);
    hbaseClient.CreateTable(TableSchema("person", {{"_doc", 1}}));
    hbaseClient.CreateTable(TableSchema("recode", {{"_doc", 1}}));
    hbaseClient.CreateTable(TableSchema("file", {{"_doc", 1}}));

    elasticSearchClient.DeleteIndex("*");
    elasticSearchClient.CreateIndex("person");
    elasticSearchClient.CreateIndex("recode");
    elasticSearchClient.CreateIndex("file");

    return 0;
#else
    return 0;
#endif

}